package com.achen.study.study1;

import java.util.concurrent.*;


/**
 * @description: CompletableFuture异步编排
 * @author: Zhaoliang Chen
 * @time: 2021/8/16
 */
public class Study2 {

/*
    现在我们来模拟一个在家煮饭的业务流程
        1、买菜
        2、煮饭
        3、切菜,备调料
        4、热锅,炒料,炒菜
        5、洗碗后才能给女朋友洗水果吃



*/

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //todo 1、在没有异步的情况下，以上流程同步执行
//        TimeInterval timer = DateUtil.timer();
//        System.out.println(buyVegetables());
//        System.out.println(cookRice());
//        System.out.println(ready());
//        System.out.println(stirFry());
//        System.out.println(washFruit());
//        System.out.println("同步执行花费秒数 : " + timer.interval() / 1000); //--------------- : 同步执行花费秒数 : 12

        //创建异步对象的四种方式，先把总结写在前面
            // CompletableFuture.runAsync(Runnable runnable)
            // CompletableFuture.runAsync(Runnable runnable, Executor executor)
            // CompletableFuture.supplyAsync(Supplier<U> supplier)
            // CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)

        // 上面的四个方法从有无返回值的方式区分可以分为两类 ：runXxxx都是没有返回结果的，supplyXxx都是可以获取返回结果的
        // 上面的四个方法从有无自定义线程池区分可以分为两类 ：没有指定Executor的方法会使用 ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行


     // 下面我们来一个简单的入门使用
//        System.out.println(Thread.currentThread().getId() + " 主线程在跑");
//        CompletableFuture.runAsync(() -> {
//            System.out.println(Thread.currentThread().getId() + " 异步线程在跑, runAsync 不支持返回值");
//        }).whenComplete((r, u) -> {
//            System.out.println(Thread.currentThread().getId() + " 异步线程在跑, runAsync 完成");
//        });
//
//        CompletableFuture.supplyAsync(() -> {
//            System.out.println(Thread.currentThread().getId() + " 异步线程在跑, supplyAsync 支持返回值");
//            return Thread.currentThread().getId();
//        }).whenComplete((r, u) -> {
//            System.out.println(Thread.currentThread().getId() + " 异步线程在跑, supplyAsync 完成,获得结果 :" + r);
//        });
//        System.out.println(Thread.currentThread().getId() + " 主线程在跑");


        //在上面的代码中，我们发现我们在异步任务的后面都跟了一个 whenComplete() 方法，当CompletableFuture的计算结果完成，或者抛出异常的时候，可以执行特定的Action。主要是下面的方法
            // 1、whenComplete(BiConsumer<? super T,? super Throwable> action)
            // 2、whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
            // 3、whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
            // 4、exceptionally(Function<Throwable,? extends T> fn)
            // 区别
                // whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务,也有可能不会是当前线程，看下面的Demo解释
                // whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。
        // 可以看到Action的类型是BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况



        //线程串行化,可别见下面的方法很多 ，其实记住三个最外面的就行，每个里面都有三个方法，分别为 ： 同步不换线程继续执行、异步切换线程执行无自定义线程池、异步切换线程执行有自定义线程池

            // thenApply  方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值
                    // public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
                    // public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
                    // public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

            // thenAccept 方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果
                    //public CompletionStage<Void> thenAccept(Consumer<? super T> action);
                    //public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
                    //public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

            // thenRun    方法：只要上面的任务执行完成，就开始执行thenRun，只是处理完任务后，执行 thenRun的后续操作
                    //public CompletionStage<Void> thenRun(Runnable action);
                    //public CompletionStage<Void> thenRunAsync(Runnable action);
                    //public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);
            //PS : Function<? super T,? extends U>
                    // T：上一个任务返回结果的类型
                    // U：当前任务的返回值类型

        //两个任务必须都完成，触发该任务
            // thenCombine：   组合两个future，获取两个future任务的返回结果，并返回当前任务的返回值
            // thenAcceptBoth：组合两个future，获取两个future任务的返回结果，然后处理任务，没有返回值。
            // runAfterBoth：  组合两个future，不需要获取future的结果，只需两个future处理完任务后，处理该任务


        //多任务组合
            // allOf(CompletableFuture<?>... cfs)  等待所有任务完成
            // anyOf(CompletableFuture<?>... cfs)  只要有一个任务完成


        //------------下面开始演示上面的代码，引入自定义线程池，演示各路API的使用------------------
        ThreadPoolExecutor customeThreadPool = new ThreadPoolExecutor(
                20,
                200,
                10,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );

        //todo 1、 创建异步任务的四种方式【有无返回值、有无自定义线程池】，跳过

        //todo 2、 针对异步的结果的处理【异步线程继续执行、异步线程交给默认线程池执行、异步线程交给自定义线程池执行】
        //2.1 异步线程继续执行? 为什么whenComplete()中的线程发生了变化，而不是执行上面业务的线程？？？
        //当代码运行到whenComplete时候，如果计算已经还没有完成，线程不切换，仍然使用计算线程来继续执行BiConsumer中的操作
        //当代码运行到whenComplete时，如果计算结果已经完成，则此时计算线程可能已经进行其他操作，因此whenComplete中的操作由运行whenComplete的线程继续执行
        System.out.println( "main start threadId : " + Thread.currentThread().getId());
        CompletableFuture<Long> exceptionally = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getId() + " 号线程在执行任务，并返回当前线程ID");
//            int flag = 10 / 0;
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Thread.currentThread().getId();
        },customeThreadPool).whenComplete((res, ex) -> {
            long currentThreadId = Thread.currentThread().getId();
            System.out.println("当前线程为" + currentThreadId + ",上一步返回的线程ID为：" + res);
        }).exceptionally(e -> {
            System.out.println("如果上面发生了异常,我可以修改最终返回值,如果没发生异常，就假装没看见过我吧");
            System.out.println("好像上面确实出错了，我给你一点出错的信息吧 " + e.getMessage());
            return Long.valueOf("0");
        });

        System.out.println("获取上面异步的响应" + exceptionally.get());
        System.out.println( "main end threadId : " + Thread.currentThread().getId());


        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(8);
        scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("延迟了几秒后才会执行哦");
            }
        }, 60, TimeUnit.SECONDS);


    }

    //买菜
    public static String buyVegetables(){
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {}

        return "花了3秒钟去买菜";
    }

    //煮饭
    public static String cookRice(){
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {}

        return "花了3秒钟去煮饭";
    }

    //切菜，备调料
    public static String ready(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {}

        return "花了1秒钟去准备调料";
    }

    //热锅,炒料,炒菜
    public static String stirFry(){
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {}

        return "花了2秒钟去炒菜";
    }

    //给女朋友洗水果吃
    public static String washFruit(){
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {}

        return "花了3秒钟洗了碗，还给女朋友洗了水果";
    }



}
